package com.lsx143.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lsx143.realtime.app.BaseApp;
import com.lsx143.realtime.common.Constants;
import com.lsx143.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.List;
import java.util.Map;

/**
 * DWM-跳出明细APP
 */
public class DWMJVApp extends BaseApp {
    public static void main(String[] args) {
        new DWMJVApp().init(
                20004,
                "DWMUserJumpDetailApp",
                1,
                "DWMUserJumpDetailApp",
                Constants.TOPIC_DWD_PAGE_LOG
        );
    }

    /**
     * DWM-跳出明细的业务逻辑实现
     *
     * @param env       flink的执行环境
     * @param srcStream 输入流
     */
    @Override
    protected void run(StreamExecutionEnvironment env, DataStreamSource<String> srcStream) {

        //测试数据
        /*srcStream = env.fromElements(
                "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000} ",
                "{\"common\":{\"mid\":\"101\"},\"page\":{\"page_id\":\"home\"},\"ts\":11000} ",
                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"home\"},\"ts\":10000}",
                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
                        "\"home\"},\"ts\":39999} ",
                "{\"common\":{\"mid\":\"102\"},\"page\":{\"page_id\":\"good_list\",\"last_page_id\":" +
                        "\"detail\"},\"ts\":50000} "
        );*/

        //1.封装数据,添加水印,根据mid分组
        KeyedStream<JSONObject, String> keyedStream = srcStream
                .map(JSON::parseObject)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((ele, ts) -> ele.getLong("ts"))
                )
                .keyBy(data -> data.getJSONObject("common").getString("mid"));

        //2.CEP模式
        /*
         * 这里使用匹配两次进入的入口
         * 1.用于匹配 连续 两次只打开主页,直接关闭的情况,这时匹配的结果是"直接跳出"
         * 2.如果匹配不上,那么在超时中,提取出来的结果就是"超时跳出"
         */
        Pattern<JSONObject, JSONObject> pattern = Pattern
                .<JSONObject>begin("login")
                //2.1 过滤出第一次访问
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) {
                        String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                        return lastPageId == null || lastPageId.length() == 0;
                    }
                })
                //2.2 过滤出连续的第二次访问
                .next("nextLogin")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject jsonObject) {
                        String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                        return lastPageId == null || lastPageId.length() == 0;
                    }
                })
                //2.3 会话必须在30s之内
                .within(Time.seconds(30));

        //3.把模式运用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        //4.获取匹配的结果
        OutputTag<String> timeOutTag = new OutputTag<String>("timeOutTag") {
        };
        @SuppressWarnings("all")
        SingleOutputStreamOperator<String> jumpOutStream = patternStream.flatSelect(
                timeOutTag,
                //4.1 超时的数据 -> "超时跳出"
                new PatternFlatTimeoutFunction<JSONObject, String>() {
                    @Override
                    public void timeout(Map<String, List<JSONObject>> map,
                                        long timeoutTimestamp,
                                        Collector<String> out) {
                        List<JSONObject> objs = map.get("login");
                        for (JSONObject obj : objs) {
                            System.out.println("超时跳出: mid=" + obj.getJSONObject("common").getString("mid"));
                            out.collect(obj.toJSONString());
                        }
                    }
                },
                //4.2 匹配的数据 -> "直接跳出"
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> map,
                                           Collector<String> out) {
                        //1.这里也是匹配"第一次登录",第二次登录的判断交给下次匹配即可
                        List<JSONObject> objs = map.get("login");
                        for (JSONObject obj : objs) {
                            System.out.println("直接跳出: mid=" + obj.getJSONObject("common").getString("mid"));
                            out.collect(obj.toJSONString());
                        }
                    }
                });
        //5.汇总"超时跳出"和"直接跳出"的数据
        DataStream<String> timeOutStream = jumpOutStream.getSideOutput(timeOutTag);
        DataStream<String> resultStream = jumpOutStream.union(timeOutStream);
        //6.输出数据到kafka
        // 测试使用
//        resultStream.print("resultStream");
        resultStream.addSink(KafkaUtil.getKafkaSink(Constants.TOPIC_DWM_JV));
    }
}
